/*
 * Copyright (c) 2012-2014 The original author or authors
 * ------------------------------------------------------
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * and Apache License v2.0 which accompanies this distribution.
 *
 * The Eclipse Public License is available at
 * http://www.eclipse.org/legal/epl-v10.html
 *
 * The Apache License v2.0 is available at
 * http://www.opensource.org/licenses/apache2.0.php
 *
 * You may elect to redistribute this code under either of these licenses.
 */
package org.dna.mqtt.moquette.messaging.spi.impl;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.OutputMessagingEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.lmax.disruptor.BatchEventProcessor;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;

/**
 * LMAX disruptor that serializes all the output message to Netty I/O.  
 * @author andrea, puran
 */
public class MessageOutputDisruptor implements EventHandler<ValueEvent> {
	private static final Logger LOG = LoggerFactory.getLogger(MessageOutputDisruptor.class);
	
	/* LMAX disruptor singleton instance, there should be one and only one outbound disruptor instance per broker instance */
	private static MessageOutputDisruptor OUTPUT_DISRUPTOR_INSTANCE;
	
	private ExecutorService m_executor;
    private BatchEventProcessor<ValueEvent> m_eventProcessor;
    private RingBuffer<ValueEvent> m_ringBuffer;
    
    /**
	 * Returns singleton inbound disruptor instance.
	 * @return the disruptor instance
	 */
	public static MessageOutputDisruptor getInstance() {
		if (OUTPUT_DISRUPTOR_INSTANCE == null) {
			OUTPUT_DISRUPTOR_INSTANCE = new MessageOutputDisruptor();
		}
		return OUTPUT_DISRUPTOR_INSTANCE;
	}
	
    public void init() {
    	//init the output ringbuffer
        m_executor = Executors.newFixedThreadPool(1);

        m_ringBuffer = new RingBuffer<ValueEvent>(ValueEvent.EVENT_FACTORY, 1024 * 32);

        SequenceBarrier barrier = m_ringBuffer.newBarrier();
        m_eventProcessor = new BatchEventProcessor<ValueEvent>(m_ringBuffer, barrier, this);
       
        //TODO in a presentation is said to don't do the followinf line!!
        m_ringBuffer.setGatingSequences(m_eventProcessor.getSequence());
        m_executor.submit(m_eventProcessor);
    }

    public void stop() {
    	LOG.debug("Shutting down output disruptor");
    	m_executor.shutdown();
    }
    
    public void disruptorPublish(OutputMessagingEvent msgEvent) {
      LOG.debug("disruptorPublish publishing event on output {}", msgEvent);
      
      long sequence = m_ringBuffer.next();
      ValueEvent event = m_ringBuffer.get(sequence);

      event.setEvent(msgEvent);
      m_ringBuffer.publish(sequence); 
  }
	
	@Override
	public void onEvent(ValueEvent event, long sequence, boolean endOfBatch) throws Exception {
		LOG.debug("Received output message event");
		MessagingEvent evt = event.getEvent();
		
        //It's always of type OutputMessagingEvent
        OutputMessagingEvent outEvent = (OutputMessagingEvent) evt;
        outEvent.getChannel().write(outEvent.getMessage());
	}
}
